package org.hyf.inspur.LessonDesin.clear.count7;

import org.hyf.inspur.LessonDesin.clear.tools.TProperties;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public  class UserPortraitReducer extends Reducer<Text, Text, NullWritable, Text> {
    private Text result = new Text();

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException{
        // PV数
        int sumPV = 0;
        // 人数
        int sumN = 0;
        // 数据
        Map<String, Integer> map = new HashMap<String, Integer>();

        // 计算总PV
        for (Text val : values) {
            // val : 用户ID， 访问次数
            // str : 用户ID, 访问次数
            String[] str = val.toString().split(TProperties.getValue("fileoutsplit"));
            sumPV = sumPV + Integer.parseInt(str[1]);
            if (map.containsKey(str[0])) {
                // 用户访问数
                map.put(str[0], map.get(str[0]) + Integer.parseInt(str[1]));
            } else {
                // 用户访问数
                map.put(str[0], Integer.parseInt(str[1]));
                // 人数
                sumN = sumN + 1;
            }
        }
        // z-score
        // 平方累加和
        double math2 = 0;
        // 均值
        double avg = sumPV / sumN;
        // 方差分子
        for (String s : map.keySet()) {
            math2 = math2 + Math.pow(map.get(s) - avg, 2);
        }
        // 标准差
        double fc = Math.sqrt(math2 / sumN);
        // 遍历数据
        for (String skey : map.keySet()) {
            // z-score标准化
            double b = (map.get(skey) -avg) / fc;
            // 数据平移
            b = b + 5;
            // 上限为10
            if (b > 10) {
                b = 10;
            }
            // 下限为0
            if (b < 0) {
                b = 0;
            }
            result = new Text(skey + TProperties.getValue("outfilesplit") + key.toString() +
                    TProperties.getValue("outfilesplit") + map.get(skey) +
                    TProperties.getValue("outfilesplit") + sumPV +
                    TProperties.getValue("outfilesplit") + b);
            context.write(NullWritable.get(), result);
        }
    }
}
